package com.adcc.automation.service.transfer;

import com.adcc.aoc.devLib.traffic.mq.configuration.ActiveMQConfiguration;
import com.adcc.aoc.devLib.traffic.mq.configuration.MQConfiguration;
import com.adcc.aoc.devLib.traffic.mq.connectionPool.ActiveMQConnectionPool;
import com.adcc.aoc.devLib.traffic.mq.connectionPool.MQConnectionPool;
import com.adcc.aoc.devLib.traffic.mq.connectionPool.MQConnectionPoolFactory;
import com.adcc.aoc.devLib.traffic.mq.transmitter.*;
import com.adcc.aoc.devLib.traffic.mq.transmitter.activemq.AckMode;
import com.adcc.aoc.devLib.traffic.mq.transmitter.activemq.ActiveMQTransmitter;
import com.adcc.automation.communication.CommunicationHandler;
import com.adcc.automation.service.configuration.ConnectionStatus;
import com.adcc.automation.service.configuration.MQProperties;
import com.adcc.automation.service.configuration.Operation;
import lombok.Getter;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

public class ActiveMqFactory implements MsgRecvListener, MQStatusListener {
    private static final Logger logger = LoggerFactory.getLogger(ActiveMqFactory.class);
    Map<String, Consumer<Message>> topicActions = new HashMap<>();
    Map<String, Consumer<Message>> queueActions = new HashMap<>();
    @Setter
    private CommunicationHandler communicationHandler;
    
    @Getter
    public MQProperties activeMqProperties;
    
    private MQTransmitter transmitter;
    
    private MQConnectionPool pool;
    
    @Setter
    private TransferListener listener;
    
    private int source;
    
    /**
     * 初始化mq
     */
    public void init(int source) {
        try {
            this.source = source;
            MQConfiguration configuration = new ActiveMQConfiguration(activeMqProperties.getUser(), activeMqProperties.getPassword(), activeMqProperties.getUrl());
            pool = MQConnectionPoolFactory.getInstance().create(configuration);
            ((ActiveMQConnectionPool) pool).setMaxConnections(activeMqProperties.getMaxConnections());
            ((ActiveMQConnectionPool) pool).setMaxActive(activeMqProperties.getMaxActiveConnections());
            ((ActiveMQConnectionPool) pool).setIdleTimeout(activeMqProperties.getIdleTimeout());
            ((ActiveMQConnectionPool) pool).setExpiryTimeout(activeMqProperties.getExpiryTimeout());
            pool.init();
            transmitter = MQTransmitterFactory.getInstance().create(configuration, pool);
            ((ActiveMQTransmitter) transmitter).setAckMode(AckMode.values()[activeMqProperties.getAckMode()]);
            //配置状态监听
            transmitter.setStatusListener(this);
            //配置队列监听
            transmitter.setQMsgListener(this, activeMqProperties.getRecvCommandQueue());
            transmitter.setTMsgListener(this, activeMqProperties.getRecvTopic());
            //动态监听queue
            createTopicAction(activeMqProperties);
            createQueueAction(activeMqProperties);
        } catch (Exception e) {
            logger.error("ActiveMqFactory init error", e);
        }
    }
    private void createTopicAction(MQProperties properties) {
        if (properties.getRecvTopic() != null && !properties.getRecvTopic().isEmpty()) {
            topicActions.put(properties.getRecvTopic(), (message) -> {
                try {
                    listener.onData(source,Operation.TOPIC,new String(message.getContent()));
                } catch (Exception e) {
                    logger.error("receive topic: {} message:{}", properties.getRecvTopic(),new String(message.getContent()), e);
                }
            });
        }
    }
    
    private void createQueueAction(MQProperties properties) {
        if (properties.getRecvCommandQueue()!= null &&!properties.getRecvCommandQueue().isEmpty()) {
            queueActions.put(properties.getRecvCommandQueue(), (message) -> {
                try {
                    listener.onData(source,Operation.COMMAND_QUEUE,new String(message.getContent()));
                } catch (Exception e) {
                    logger.error("receive queue: {} message:{}", properties.getRecvCommandQueue(),new String(message.getContent()), e);
                }
            });
        }
    }
    
    @Override
    public void onState(MQStatus mqStatus, Map<String, String> map) {
        logger.info(String.format("onState mqStatus: %s, map:%s", mqStatus, map));
        if (mqStatus == MQStatus.CONNECTED) {
            listener.onState(ConnectionStatus.CONNECTED);
        } else if (mqStatus == MQStatus.DISCONNECTED) {
            listener.onState(ConnectionStatus.DISCONNECTED);
        } else if (mqStatus == MQStatus.CONNECTING) {
            listener.onState(ConnectionStatus.CONNECTING);
        }
    }
    
    @Override
    public void onMsg(String s, Message message, int i) {
        logger.info("receive mq onMsg: {}i:{}", new String(message.getContent()), i);
    }
    
    @Override
    public void onMsg(String s, Message message, Map<String, String> map, int i) {
        if (i == 0) {//消息
            if (message.getLength() > 0) {
                Consumer<Message> handler = queueActions.get(s);
                if (handler != null) {
                    handler.accept(message);
                } else {
                    logger.info("receive queue: {} message:{}", s, new String(message.getContent()));
                }
            }
        } else {//心跳
            if (message.getLength() > 0) {
                Consumer<Message> handler = topicActions.get(s);
                if (handler != null) {
                    handler.accept(message);
                } else {
                    logger.info("receive topic: {} message:{}", s, new String(message.getContent()));
                }
            }
        }
    }
    
    public void start() throws Exception {
        try {
            transmitter.startAsync();
        } catch (Exception e) {
            logger.error("start mq error", e);
            throw e;
        }
    }
    
    public void stop() throws Exception {
        try {
            pool.dispose();
            transmitter.stopAsync();
        } catch (Exception e) {
            logger.error("stop mq error", e);
            throw e;
        }
    }
    
    /**
     * 配置参数
     */
    public ActiveMqFactory setConfig(Map<String, Object> map) throws Exception {
        try {
            activeMqProperties = new MQProperties(map);
            return this;
        } catch (Exception e) {
            logger.error("setConfig error", e);
            throw e;
        }
    }
    
    public void send(String queue, String data) throws Exception {
        try {
            if(queue.equals(activeMqProperties.getSendTopic())){
                transmitter.sendTopic(queue, new Message(data.getBytes()), activeMqProperties.getTtl());
                logger.info("send topic: {} message:{}", queue, data);
            }else if(queue.equals(activeMqProperties.getSendCommandQueue()) || queue.equals(activeMqProperties.getFplQueue())){
                if(activeMqProperties.isTextTransfer()){
                    transmitter.sendQueue(queue, new Message(data), activeMqProperties.getTtl());
                    logger.info("send queue by text: {} message:{}", queue, data);
                }else{
                    transmitter.sendQueue(queue, new Message(data.getBytes()), activeMqProperties.getTtl());
                    logger.info("send queue by byte: {} message:{}", queue, new String(data.getBytes()));
                }
            }
        } catch (Exception e) {
            logger.error("send mq error", e);
            throw e;
        }
    }
}
